package com.kingsoft.dc.khaos.module.spark.preprocess.specific.quality

import java.text.SimpleDateFormat
import java.util
import java.util.Date

import com.alibaba.fastjson.JSON
import com.kingsoft.dc.khaos.KhaosContext
import com.kingsoft.dc.khaos.extender.meta.model.check.{DmCheck, MQCheck}
import com.kingsoft.dc.khaos.innertype.Schema
import com.kingsoft.dc.khaos.metadata.{Dependency, KhaosStructField}
import com.kingsoft.dc.khaos.module.spark.constants.{CheckRuleConstants, SchedulerConstants}
import com.kingsoft.dc.khaos.module.spark.metadata.preprocess.specific.quality.{TechCheckFieldInfo, TechCheckInfo}
import com.kingsoft.dc.khaos.module.spark.preprocess.transform.TransformStrategy
import com.kingsoft.dc.khaos.module.spark.util._
import com.kingsoft.dc.khaos.util.Logging
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.encoders.RowEncoder
import org.apache.spark.sql.functions.{col, concat_ws}
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Column, DataFrame, Row}
import org.json4s.DefaultFormats
import org.json4s.jackson.JsonMethods.{compact, parse, render}

import scala.collection.JavaConversions._
import scala.collection.mutable.ArrayBuffer
import scala.util.control.Breaks.{break, breakable}

class TechCheck extends TransformStrategy  with Logging with Serializable {

  private var techCheckInfo: TechCheckInfo = null
  private var mqCheckInfo: MQCheck = null

  override def exec(kc: KhaosContext,
                    module_id: String,
                    config: String,
                    dependences: Seq[Dependency],
                    targets:Seq[Dependency]): Seq[(String,DataFrame)] = {
//    log.info(s"tech check start ......")
//    log.info(s"tech check config json is: ${config}")

    init(kc, config)
    val runEnv = kc.conf.getString(SchedulerConstants.RUN_ENV)

    //获取上游信息
    val (schema, df) = kc.structData[DataFrame](dependences.head.getEdge())
//    log.info(s"check source dataFrame partitions size is: ${df.rdd.partitions.size}")
//    log.info(s"check source dataFrame schema is: ${df.printSchema}")
//    log.info(s"check source dataFrame is: ${df.show(100,false)}")

    //将df字段名统一转换为小写
    val lowerCaseDf = renameDF(df)
    //将df字段中的timestamp类型统一转为String
    val dataFrame = timestamp2String(lowerCaseDf)
//    log.info(s"check dataFrame partitions size is: ${dataFrame.rdd.partitions.size}")
//    log.info(s"check dataFrame schema is: ${dataFrame.printSchema}")
//    log.info(s"check dataFrame is: ${dataFrame.show(100,false)}")

    val normalDataDF = recordNormalData(dataFrame, techCheckInfo.extract_fields)

    val resultDFToMQ = recordRejectAndWarnInfoToMQ(kc, dataFrame, techCheckInfo.extract_fields, dataFrame.count)
//    log.info(s"resultDFToMQ schema is: ${resultDFToMQ.printSchema}")
//    log.info(s"resultDFToMQ is: ${resultDFToMQ.show(100,false)}")
    if (!resultDFToMQ.rdd.isEmpty() && runEnv.equalsIgnoreCase("online")) {
      checkSinkMQ(kc, resultDFToMQ)
    }

    var resultDFToES = recordRejectAndWarnInfoToES(kc, dataFrame, techCheckInfo.extract_fields)
//    log.info(s"resultDFToES schema is: ${resultDFToES.printSchema}")
//    log.info(s"resultDFToES is: ${resultDFToES.show(100,false)}")

    var replacedDF = replaceWarnData(normalDataDF, techCheckInfo.extract_fields)
//    log.info(s"replacedDF schema is: ${replacedDF.printSchema}")
//    log.info(s"replacedDF is: ${replacedDF.show(100,false)}")

    //默认检核任务不中断
    kc.conf.set("stop_check", "false")
    //输出到ES的DF
    if (runEnv.equalsIgnoreCase("online")) {
      //当检核上报条数大于用户指定最大值时，检核任务中断
      if (techCheckInfo.check_sink_nums.nonEmpty && resultDFToES.count > techCheckInfo.check_sink_nums.get) {
        kc.conf.set("stop_check", "true")
        resultDFToES = resultDFToES.limit(techCheckInfo.check_sink_nums.get)
        addResult(targets.filter(_.targetSeq.equals("1")).head, resultDFToES)
      } else {
        addResult(targets.filter(_.targetSeq.equals("1")).head, resultDFToES)
      }
    } else {
      log.info(s"测试态数据检核上报功能关闭！")
      addResult(targets.filter(_.targetSeq.equals("1")).head, emptyESDf(kc))
    }

    //将检核后的df字段中修改过的String类型还原为timestamp
    replacedDF = transformDfSchema(lowerCaseDf, replacedDF)
//    log.info(s"replacedDF schema is: ${replacedDF.printSchema}")
//    log.info(s"replacedDF is: ${replacedDF.show(100,false)}")

    //将检核后的df字段类型由String类型转为预期类型
    replacedDF = convertDFType(kc, replacedDF, techCheckInfo.extract_fields)

//    log.info(s"transform replacedDF partitions size is: ${replacedDF.rdd.partitions.size}")
//    log.info(s"transform replacedDF schema is: ${replacedDF.printSchema}")
//    log.info(s"transform replacedDF is: ${replacedDF.show(100,false)}")

//    val filterExpr = kc.conf.getString("source.cos.filter")
//    if (filterExpr != null && !filterExpr.trim.equals("")) {
//      replacedDF = replacedDF.filter(filterExpr)
//    }
//
//    log.info(s"filter replacedDF partitions size is: ${replacedDF.rdd.partitions.size}")
//    log.info(s"filter replacedDF schema is: ${replacedDF.printSchema}")
//    log.info(s"filter replacedDF is: ${replacedDF.show(100,false)}")

    addResult(targets.filter(_.targetSeq.equals("0")).head, replacedDF)
  }

  //21202F2938212B3E22272626252E434D
  override def schema(kc: KhaosContext,
                      module_id: String,
                      config: String,
                      dependencies: Seq[Dependency]): Schema = {

    val schema = if (techCheckInfo.extract_fields.isEmpty) {
      kc.schemaChannel.getSchema(dependencies.head.getSource())
    } else {
      val schemaList = techCheckInfo.extract_fields.map { optInfo =>
        KhaosStructField(optInfo.field, optInfo.data_type)
      }
      new Schema(schemaList)
    }
    schema
  }

  /** 初始化参数 */
  def init(kc: KhaosContext, config: String): Unit = {
    if (true == false) {
      val str = "21202F2938212B3E22272626252E434D"
    }
    //获取配置信息
    implicit val formats = DefaultFormats
    techCheckInfo = parse(config, true).extract[TechCheckInfo]

    //获取检核上报MQ配置
    mqCheckInfo = CheckUtils.getMQCheckInfo(kc, techCheckInfo.extender.check.clazz)
  }

  def recordNormalData(data: DataFrame, extractFields: List[TechCheckFieldInfo]): DataFrame = {

    val df: DataFrame = data

    val schemaFields = df.schema.fields
    val encoder = RowEncoder(df.schema)

    val normalDataDF: DataFrame = df.flatMap(line => {
      val normalDataArray = ArrayBuffer[Row]()
      var normalFlag = true

      breakable(

        extractFields.foreach(ef => {
          val fieldName = ef.field.toLowerCase
          val onOff = ef.option.tech_check.on_off
//          val ruleOptions = ef.option.tech_check.rule_options
          if(onOff) {
            //记录正常的数据记录
            for (rule_type <- CheckRuleConstants.ALL_RULE_TYPE) {
              rule_type match {
                case "CONTENT_LENGTH" => {
                  if (ef.data_type.equalsIgnoreCase("STRING")) {
                    if (ef.length.nonEmpty && TechCheckRuleUtils.strOverLength(line.getAs[String](fieldName), ef.length.get)) {
                      normalFlag = false
                      break()
                    }
                  } else if (ef.data_type.equalsIgnoreCase("NUMBER")) {
                    if (line.getAs(fieldName) != null) {
                      if (ef.length.nonEmpty && TechCheckRuleUtils.intOverLength(line.getAs(fieldName).toString, ef.length.get)) {
                        normalFlag = false
                        break()
                      }
                    } else {
                      if (ef.length.nonEmpty && TechCheckRuleUtils.intOverLength(line.getAs(fieldName), ef.length.get)) {
                        normalFlag = false
                        break()
                      }
                    }
                  } else if (ef.data_type.equalsIgnoreCase("DECIMAL")) {
                    if (line.getAs(fieldName) != null) {
                      if (ef.length.nonEmpty && TechCheckRuleUtils.decimalOverLength(line.getAs(fieldName).toString, ef.length.get)) {
                        normalFlag = false
                        break()
                      }
                    } else {
                      if (ef.length.nonEmpty && TechCheckRuleUtils.decimalOverLength(line.getAs(fieldName), ef.length.get)) {
                        normalFlag = false
                        break()
                      }
                    }
                  }
                }
                case "CONTENT_FORMAT" => {
                  if (ef.data_type.equalsIgnoreCase("DATE")) {
                    if (line.getAs(fieldName) != null) {
                      if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName).toString, CheckRuleConstants.DATE_FORMAT_STYLE_ARRAY)) {
                        normalFlag = false
                        break()
                      }
                    } else {
                      if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName), CheckRuleConstants.DATE_FORMAT_STYLE_ARRAY)) {
                        normalFlag = false
                        break()
                      }
                    }
                  } else if (ef.data_type.equalsIgnoreCase("TIME")) {
                    if (line.getAs(fieldName) != null) {
                      if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName).toString, CheckRuleConstants.TIME_FORMAT_STYLE_ARRAY)) {
                        normalFlag = false
                        break()
                      }
                    } else {
                      if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName), CheckRuleConstants.TIME_FORMAT_STYLE_ARRAY)) {
                        normalFlag = false
                        break()
                      }
                    }
                  } else if (ef.data_type.equalsIgnoreCase("DATETIME")) {
                    if (line.getAs(fieldName) != null) {
                      if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName).toString, CheckRuleConstants.DATETIME_FORMAT_STYLE_ARRAY)) {
                        normalFlag = false
                        break()
                      }
                    } else {
                      if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName), CheckRuleConstants.DATETIME_FORMAT_STYLE_ARRAY)) {
                        normalFlag = false
                        break()
                      }
                    }
                  }
                }
                case _ =>
              }
            }
          }
        })
      )
      if (normalFlag) {
        normalDataArray += line
      }
      normalDataArray
    })(encoder)
    normalDataDF
  }

  /** 记录拒绝信息 */
  def recordRejectAndWarnInfoToMQ(kc: KhaosContext, data: DataFrame, extractFields: List[TechCheckFieldInfo], totalNum: Long): DataFrame = {
    //任务级别参数
    val taskFlowNum = s"${kc.conf.getString(SchedulerConstants.JOB_INSTANCE_ID)}-${kc.conf.getString(SchedulerConstants.EXEC_TRIED_TIMES)}"
    val taskId = kc.conf.getString(SchedulerConstants.JOB_ID)
    val taskName = kc.conf.getString(SchedulerConstants.JOB_NAME)
    val taskType = "DATA_INTEGRATION"
    val businessTime = s"${string2Date(kc.conf.getString(SchedulerConstants.BIZ_DATE))} ${kc.conf.getString(SchedulerConstants.BIZ_TIME)}"
    val taskSubmitTime = timeStamp2Date(kc.conf.getString(SchedulerConstants.SUBMIT_TIME))
    val batchNum = kc.conf.getString(SchedulerConstants.BATCH_NO)
    val taskProjectId = kc.conf.getString(SchedulerConstants.PROJECT_ID)
    val taskProjectName = kc.conf.getString(SchedulerConstants.PROJECT_NAME)

    val checkParams = compact(render(techCheckInfo.extender.check.params))
    val dataSourceId = JSON.parseObject(checkParams).get("ds_id").toString
    val dataSourceName = JSON.parseObject(checkParams).get("ds_name").toString
    val dbName = techCheckInfo.db_name
    val tableName = techCheckInfo.table_name

    val df: DataFrame = data

    val schema = StructType(Seq(StructField("reject_info_mq", StringType, true), StructField("rule_id", StringType, true), StructField("column_name", StringType, true), StructField("rule_status", StringType, true)))
    val schemaFields = df.schema.fields
    val encoder = RowEncoder(schema)

    val rejectDF: DataFrame = df.flatMap(line => {
      val rejectArray = ArrayBuffer[Row]()

      extractFields.foreach(ef => {
        val fieldName = ef.field.toLowerCase
        val onOff = ef.option.tech_check.on_off
//        val ruleOptions = ef.option.tech_check.rule_options
        if(onOff) {
          //记录拒绝信息
          for (rule_type <- CheckRuleConstants.ALL_RULE_TYPE) {
            rule_type match {
              case "CONTENT_LENGTH" => {
                if (ef.data_type.equalsIgnoreCase("STRING")) {
                  if (ef.length.nonEmpty && TechCheckRuleUtils.strOverLength(line.getAs[String](fieldName), ef.length.get)) {
                    var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|strOverLength")
                    row = Row.merge(row, Row(CheckRuleConstants.STRING_CONTENT_LENGTH_RULE_ID.toString))
                    row = Row.merge(row, Row(ef.field))
                    row = Row.merge(row, Row("reject"))
                    rejectArray += row
                  } else {
                    var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|strOverLength")
                    row = Row.merge(row, Row(CheckRuleConstants.STRING_CONTENT_LENGTH_RULE_ID.toString))
                    row = Row.merge(row, Row(ef.field))
                    row = Row.merge(row, Row("pass"))
                    rejectArray += row
                  }
                } else if (ef.data_type.equalsIgnoreCase("NUMBER")) {
                  if (line.getAs(fieldName) != null) {
                    if (ef.length.nonEmpty && TechCheckRuleUtils.intOverLength(line.getAs(fieldName).toString, ef.length.get)) {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|intOverLength")
                      row = Row.merge(row, Row(CheckRuleConstants.NUMBER_CONTENT_LENGTH_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("reject"))
                      rejectArray += row
                    } else {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|intOverLength")
                      row = Row.merge(row, Row(CheckRuleConstants.NUMBER_CONTENT_LENGTH_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("pass"))
                      rejectArray += row
                    }
                  } else {
                    if (ef.length.nonEmpty && TechCheckRuleUtils.intOverLength(line.getAs(fieldName), ef.length.get)) {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|intOverLength")
                      row = Row.merge(row, Row(CheckRuleConstants.NUMBER_CONTENT_LENGTH_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("reject"))
                      rejectArray += row
                    } else {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|intOverLength")
                      row = Row.merge(row, Row(CheckRuleConstants.NUMBER_CONTENT_LENGTH_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("pass"))
                      rejectArray += row
                    }
                  }
                } else if (ef.data_type.equalsIgnoreCase("DECIMAL")) {
                  if (line.getAs(fieldName) != null) {
                    if (ef.length.nonEmpty && TechCheckRuleUtils.decimalOverLength(line.getAs(fieldName).toString, ef.length.get)) {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|decimalOverLength")
                      row = Row.merge(row, Row(CheckRuleConstants.DECIMAL_CONTENT_LENGTH_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("reject"))
                      rejectArray += row
                    } else {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|decimalOverLength")
                      row = Row.merge(row, Row(CheckRuleConstants.DECIMAL_CONTENT_LENGTH_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("pass"))
                      rejectArray += row
                    }
                  } else {
                    if (ef.length.nonEmpty && TechCheckRuleUtils.decimalOverLength(line.getAs(fieldName), ef.length.get)) {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|decimalOverLength")
                      row = Row.merge(row, Row(CheckRuleConstants.DECIMAL_CONTENT_LENGTH_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("reject"))
                      rejectArray += row
                    } else {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|decimalOverLength")
                      row = Row.merge(row, Row(CheckRuleConstants.DECIMAL_CONTENT_LENGTH_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("pass"))
                      rejectArray += row
                    }
                  }
                }
              }
              case "CONTENT_FORMAT" => {
                if (ef.data_type.equalsIgnoreCase("DATE")) {
                  if (line.getAs(fieldName) != null) {
                    if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName).toString, CheckRuleConstants.DATE_FORMAT_STYLE_ARRAY)) {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|isValidDate")
                      row = Row.merge(row, Row(CheckRuleConstants.DATE_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("reject"))
                      rejectArray += row
                    } else {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|isValidDate")
                      row = Row.merge(row, Row(CheckRuleConstants.DATE_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("pass"))
                      rejectArray += row
                    }
                  } else {
                    if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName), CheckRuleConstants.DATE_FORMAT_STYLE_ARRAY)) {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|isValidDate")
                      row = Row.merge(row, Row(CheckRuleConstants.DATE_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("reject"))
                      rejectArray += row
                    } else {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|isValidDate")
                      row = Row.merge(row, Row(CheckRuleConstants.DATE_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("pass"))
                      rejectArray += row
                    }
                  }
                } else if (ef.data_type.equalsIgnoreCase("TIME")) {
                  if (line.getAs(fieldName) != null) {
                    if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName).toString, CheckRuleConstants.TIME_FORMAT_STYLE_ARRAY)) {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|isValidTime")
                      row = Row.merge(row, Row(CheckRuleConstants.TIME_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("reject"))
                      rejectArray += row
                    } else {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|isValidTime")
                      row = Row.merge(row, Row(CheckRuleConstants.TIME_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("pass"))
                      rejectArray += row
                    }
                  } else {
                    if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName), CheckRuleConstants.TIME_FORMAT_STYLE_ARRAY)) {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|isValidTime")
                      row = Row.merge(row, Row(CheckRuleConstants.TIME_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("reject"))
                      rejectArray += row
                    } else {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|isValidTime")
                      row = Row.merge(row, Row(CheckRuleConstants.TIME_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("pass"))
                      rejectArray += row
                    }
                  }
                } else if (ef.data_type.equalsIgnoreCase("DATETIME")) {
                  if (line.getAs(fieldName) != null) {
                    if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName).toString, CheckRuleConstants.DATETIME_FORMAT_STYLE_ARRAY)) {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|isValidDateTime")
                      row = Row.merge(row, Row(CheckRuleConstants.DATETIME_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("reject"))
                      rejectArray += row
                    } else {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|isValidDateTime")
                      row = Row.merge(row, Row(CheckRuleConstants.DATETIME_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("pass"))
                      rejectArray += row
                    }
                  } else {
                    if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName), CheckRuleConstants.DATETIME_FORMAT_STYLE_ARRAY)) {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|isValidDateTime")
                      row = Row.merge(row, Row(CheckRuleConstants.DATETIME_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("reject"))
                      rejectArray += row
                    } else {
                      var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|isValidDateTime")
                      row = Row.merge(row, Row(CheckRuleConstants.DATETIME_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(ef.field))
                      row = Row.merge(row, Row("pass"))
                      rejectArray += row
                    }
                  }
                }
              }
              case "MESSY_CODE" => {
                if (ef.data_type.equalsIgnoreCase("STRING")) {
                  if (TechCheckRuleUtils.isMessyCode(line.getAs[String](fieldName))) {
                    var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|MESSY_CODE")
                    row = Row.merge(row, Row(CheckRuleConstants.MESSY_CODE_RULE_ID.toString))
                    row = Row.merge(row, Row(ef.field))
                    row = Row.merge(row, Row("warn"))
                    rejectArray += row
                  } else {
                    var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|MESSY_CODE")
                    row = Row.merge(row, Row(CheckRuleConstants.MESSY_CODE_RULE_ID.toString))
                    row = Row.merge(row, Row(ef.field))
                    row = Row.merge(row, Row("pass"))
                    rejectArray += row
                  }
                }
              }
              case "ESCAPE_CHAR" => {
                if (ef.data_type.equalsIgnoreCase("STRING")) {
                  if (TechCheckRuleUtils.isEscape(line.getAs[String](fieldName))) {
                    var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|ESCAPE_CHAR")
                    row = Row.merge(row, Row(CheckRuleConstants.ESCAPE_CHAR_RULE_ID.toString))
                    row = Row.merge(row, Row(ef.field))
                    row = Row.merge(row, Row("warn"))
                    rejectArray += row
                  } else {
                    var row: Row = Row(s"$tableName|@|$businessTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|ESCAPE_CHAR")
                    row = Row.merge(row, Row(CheckRuleConstants.ESCAPE_CHAR_RULE_ID.toString))
                    row = Row.merge(row, Row(ef.field))
                    row = Row.merge(row, Row("pass"))
                    rejectArray += row
                  }
                }
              }
              case _ =>
            }
          }
        }
      })
      rejectArray
    })(encoder)

    val countSchema = StructType(Seq(StructField("count_info_mq", StringType, true)))
    var countDF = rejectDF.groupBy("rule_id","column_name","rule_status").count()

    val countEncoder = RowEncoder(countSchema)
    countDF = countDF.flatMap(line => {
      val rowArray = ArrayBuffer[Row]()
      if (!line.getAs("rule_status").equals("pass")) {
        val json = s"""{"taskFlowNum":"$taskFlowNum|$dataSourceName|$dbName|$tableName|${line.getAs("column_name")}","taskId":"$taskId","taskName":"$taskName","taskType":"$taskType","businessTime":"$businessTime","taskSubmitTime":"$taskSubmitTime","batchNum":"$batchNum","ruleId":"${line.getAs("rule_id")}","projectId":"$taskProjectId","projectName":"$taskProjectName","dataSourceId":"$dataSourceId","dataSourceName":"$dataSourceName","dbName":"$dbName","tableName":"$tableName","columnName":"${line.getAs("column_name")}","errNum":"${line.getAs("count")}","countNum":"$totalNum","isStop":"false","testRun":"false"}""".stripMargin
        rowArray += Row(json)
      } else if (line.getAs("rule_status").equals("pass") && (line.getAs("count") == totalNum)) {
        val json = s"""{"taskFlowNum":"$taskFlowNum|$dataSourceName|$dbName|$tableName|${line.getAs("column_name")}","taskId":"$taskId","taskName":"$taskName","taskType":"$taskType","businessTime":"$businessTime","taskSubmitTime":"$taskSubmitTime","batchNum":"$batchNum","ruleId":"${line.getAs("rule_id")}","projectId":"$taskProjectId","projectName":"$taskProjectName","dataSourceId":"$dataSourceId","dataSourceName":"$dataSourceName","dbName":"$dbName","tableName":"$tableName","columnName":"${line.getAs("column_name")}","errNum":"0","countNum":"$totalNum","isStop":"false","testRun":"false"}""".stripMargin
        rowArray += Row(json)
      }
      rowArray
    })(countEncoder)
    countDF
  }

  /** 记录拒绝信息 */
  def recordRejectAndWarnInfoToES(kc: KhaosContext, data: DataFrame, extractFields: List[TechCheckFieldInfo]): DataFrame = {
    //任务级别参数
    val taskFlowNum = s"${kc.conf.getString(SchedulerConstants.JOB_INSTANCE_ID)}-${kc.conf.getString(SchedulerConstants.EXEC_TRIED_TIMES)}"
    val taskId = kc.conf.getString(SchedulerConstants.JOB_ID)
    val taskName = kc.conf.getString(SchedulerConstants.JOB_NAME)
    val taskSubmitTime = timeStamp2Date(kc.conf.getString(SchedulerConstants.SUBMIT_TIME))
    val tenantId = kc.conf.getString(SchedulerConstants.TENANT_ID)

    val checkParams = compact(render(techCheckInfo.extender.check.params))
    val dataSourceName = JSON.parseObject(checkParams).get("ds_name").toString
    val dbName = techCheckInfo.db_name
    val tableName = techCheckInfo.table_name

    val df: DataFrame = data

    val schema = StructType(Seq(StructField("taskFlowNum", StringType, true), StructField("ruleId", StringType, true),
      StructField("taskSubmitTime", StringType, true), StructField("tenantId", StringType, true),
      StructField("dataDetaile", StringType, true), StructField("errorDescribe", ArrayType(StringType), true)))
    val schemaFields = df.schema.fields
    val encoder = RowEncoder(schema)

    val rejectDF: DataFrame = df.flatMap(line => {
      val rejectArray = ArrayBuffer[Row]()

      extractFields.foreach(ef => {
        val fieldName = ef.field.toLowerCase
        val onOff = ef.option.tech_check.on_off
//        val ruleOptions = ef.option.tech_check.rule_options
        if (onOff) {
          //记录拒绝信息
          for (rule_type <- CheckRuleConstants.ALL_RULE_TYPE) {
            rule_type match {
              case "CONTENT_LENGTH" => {
                if (ef.data_type.equalsIgnoreCase("STRING")) {
                  if (ef.length.nonEmpty && TechCheckRuleUtils.strOverLength(line.getAs[String](fieldName), ef.length.get)) {
                    var row: Row = Row(s"$taskFlowNum|$dataSourceName|$dbName|$tableName|${ef.field}")
                    row = Row.merge(row, Row(CheckRuleConstants.STRING_CONTENT_LENGTH_RULE_ID.toString))
                    row = Row.merge(row, Row(taskSubmitTime))
                    row = Row.merge(row, Row(tenantId))
                    row = Row.merge(row, Row(line.mkString("|")))
                    row = Row.merge(row, Row(Array(s"$dbName|@|$tableName|@|$taskSubmitTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|不符合内容长度检查规则")))
                    rejectArray += row
                  }
                } else if (ef.data_type.equalsIgnoreCase("NUMBER")) {
                  if (line.getAs(fieldName) != null) {
                    if (ef.length.nonEmpty && TechCheckRuleUtils.intOverLength(line.getAs(fieldName).toString, ef.length.get)) {
                      var row: Row = Row(s"$taskFlowNum|$dataSourceName|$dbName|$tableName|${ef.field}")
                      row = Row.merge(row, Row(CheckRuleConstants.NUMBER_CONTENT_LENGTH_RULE_ID.toString))
                      row = Row.merge(row, Row(taskSubmitTime))
                      row = Row.merge(row, Row(tenantId))
                      row = Row.merge(row, Row(line.mkString("|")))
                      row = Row.merge(row, Row(Array(s"$dbName|@|$tableName|@|$taskSubmitTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|不符合内容长度检查规则")))
                      rejectArray += row
                    }
                  } else {
                    if (ef.length.nonEmpty && TechCheckRuleUtils.intOverLength(line.getAs(fieldName), ef.length.get)) {
                      var row: Row = Row(s"$taskFlowNum|$dataSourceName|$dbName|$tableName|${ef.field}")
                      row = Row.merge(row, Row(CheckRuleConstants.NUMBER_CONTENT_LENGTH_RULE_ID.toString))
                      row = Row.merge(row, Row(taskSubmitTime))
                      row = Row.merge(row, Row(tenantId))
                      row = Row.merge(row, Row(line.mkString("|")))
                      row = Row.merge(row, Row(Array(s"$dbName|@|$tableName|@|$taskSubmitTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|不符合内容长度检查规则")))
                      rejectArray += row
                    }
                  }
                } else if (ef.data_type.equalsIgnoreCase("DECIMAL")) {
                  if (line.getAs(fieldName) != null) {
                    if (ef.length.nonEmpty && TechCheckRuleUtils.decimalOverLength(line.getAs(fieldName).toString, ef.length.get)) {
                      var row: Row = Row(s"$taskFlowNum|$dataSourceName|$dbName|$tableName|${ef.field}")
                      row = Row.merge(row, Row(CheckRuleConstants.DECIMAL_CONTENT_LENGTH_RULE_ID.toString))
                      row = Row.merge(row, Row(taskSubmitTime))
                      row = Row.merge(row, Row(tenantId))
                      row = Row.merge(row, Row(line.mkString("|")))
                      row = Row.merge(row, Row(Array(s"$dbName|@|$tableName|@|$taskSubmitTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|不符合内容长度检查规则")))
                      rejectArray += row
                    }
                  } else {
                    if (ef.length.nonEmpty && TechCheckRuleUtils.decimalOverLength(line.getAs(fieldName), ef.length.get)) {
                      var row: Row = Row(s"$taskFlowNum|$dataSourceName|$dbName|$tableName|${ef.field}")
                      row = Row.merge(row, Row(CheckRuleConstants.DECIMAL_CONTENT_LENGTH_RULE_ID.toString))
                      row = Row.merge(row, Row(taskSubmitTime))
                      row = Row.merge(row, Row(tenantId))
                      row = Row.merge(row, Row(line.mkString("|")))
                      row = Row.merge(row, Row(Array(s"$dbName|@|$tableName|@|$taskSubmitTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|不符合内容长度检查规则")))
                      rejectArray += row
                    }
                  }
                }
              }
              case "CONTENT_FORMAT" => {
                if (ef.data_type.equalsIgnoreCase("DATE")) {
                  if (line.getAs(fieldName) != null) {
                    if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName).toString, CheckRuleConstants.DATE_FORMAT_STYLE_ARRAY)) {
                      var row: Row = Row(s"$taskFlowNum|$dataSourceName|$dbName|$tableName|${ef.field}")
                      row = Row.merge(row, Row(CheckRuleConstants.DATE_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(taskSubmitTime))
                      row = Row.merge(row, Row(tenantId))
                      row = Row.merge(row, Row(line.mkString("|")))
                      row = Row.merge(row, Row(Array(s"$dbName|@|$tableName|@|$taskSubmitTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|不符合内容格式检查规则")))
                      rejectArray += row
                    }
                  } else {
                    if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName), CheckRuleConstants.DATE_FORMAT_STYLE_ARRAY)) {
                      var row: Row = Row(s"$taskFlowNum|$dataSourceName|$dbName|$tableName|${ef.field}")
                      row = Row.merge(row, Row(CheckRuleConstants.DATE_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(taskSubmitTime))
                      row = Row.merge(row, Row(tenantId))
                      row = Row.merge(row, Row(line.mkString("|")))
                      row = Row.merge(row, Row(Array(s"$dbName|@|$tableName|@|$taskSubmitTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|不符合内容格式检查规则")))
                      rejectArray += row
                    }
                  }
                } else if (ef.data_type.equalsIgnoreCase("TIME")) {
                  if (line.getAs(fieldName) != null) {
                    if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName).toString, CheckRuleConstants.TIME_FORMAT_STYLE_ARRAY)) {
                      var row: Row = Row(s"$taskFlowNum|$dataSourceName|$dbName|$tableName|${ef.field}")
                      row = Row.merge(row, Row(CheckRuleConstants.TIME_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(taskSubmitTime))
                      row = Row.merge(row, Row(tenantId))
                      row = Row.merge(row, Row(line.mkString("|")))
                      row = Row.merge(row, Row(Array(s"$dbName|@|$tableName|@|$taskSubmitTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|不符合内容格式检查规则")))
                      rejectArray += row
                    }
                  } else {
                    if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName), CheckRuleConstants.TIME_FORMAT_STYLE_ARRAY)) {
                      var row: Row = Row(s"$taskFlowNum|$dataSourceName|$dbName|$tableName|${ef.field}")
                      row = Row.merge(row, Row(CheckRuleConstants.TIME_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(taskSubmitTime))
                      row = Row.merge(row, Row(tenantId))
                      row = Row.merge(row, Row(line.mkString("|")))
                      row = Row.merge(row, Row(Array(s"$dbName|@|$tableName|@|$taskSubmitTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|不符合内容格式检查规则")))
                      rejectArray += row
                    }
                  }
                } else if (ef.data_type.equalsIgnoreCase("DATETIME")) {
                  if (line.getAs(fieldName) != null) {
                    if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName).toString, CheckRuleConstants.DATETIME_FORMAT_STYLE_ARRAY)) {
                      var row: Row = Row(s"$taskFlowNum|$dataSourceName|$dbName|$tableName|${ef.field}")
                      row = Row.merge(row, Row(CheckRuleConstants.DATETIME_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(taskSubmitTime))
                      row = Row.merge(row, Row(tenantId))
                      row = Row.merge(row, Row(line.mkString("|")))
                      row = Row.merge(row, Row(Array(s"$dbName|@|$tableName|@|$taskSubmitTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|不符合内容格式检查规则")))
                      rejectArray += row
                    }
                  } else {
                    if (!TechCheckRuleUtils.validFormatArray(line.getAs(fieldName), CheckRuleConstants.DATETIME_FORMAT_STYLE_ARRAY)) {
                      var row: Row = Row(s"$taskFlowNum|$dataSourceName|$dbName|$tableName|${ef.field}")
                      row = Row.merge(row, Row(CheckRuleConstants.DATETIME_CONTENT_FORMAT_RULE_ID.toString))
                      row = Row.merge(row, Row(taskSubmitTime))
                      row = Row.merge(row, Row(tenantId))
                      row = Row.merge(row, Row(line.mkString("|")))
                      row = Row.merge(row, Row(Array(s"$dbName|@|$tableName|@|$taskSubmitTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|不符合内容格式检查规则")))
                      rejectArray += row
                    }
                  }
                }
              }
//              case "MESSY_CODE" => {
//                if (ef.data_type.equalsIgnoreCase("STRING")) {
//                  if (TechCheckRuleUtils.isMessyCode(line.getAs[String](fieldName))) {
//                    var row: Row = Row(s"$taskFlowNum|$dataSourceId|$dbName|$tableName|${ef.field}")
//                    row = Row.merge(row, Row(CheckRuleConstants.MESSY_CODE_RULE_ID.toString))
//                    row = Row.merge(row, Row(taskSubmitTime))
//                    row = Row.merge(row, Row(tenantId))
//                    row = Row.merge(row, Row(line.mkString("|")))
//                    row = Row.merge(row, Row(Array(s"$dbName|@|$tableName|@|$taskSubmitTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|不符合乱码检查规则")))
//                    rejectArray += row
//                  }
//                }
//              }
//              case "ESCAPE_CHAR" => {
//                if (ef.data_type.equalsIgnoreCase("STRING")) {
//                  if (TechCheckRuleUtils.isEscape(line.getAs[String](fieldName))) {
//                    var row: Row = Row(s"$taskFlowNum|$dataSourceId|$dbName|$tableName|${ef.field}")
//                    row = Row.merge(row, Row(CheckRuleConstants.ESCAPE_CHAR_RULE_ID.toString))
//                    row = Row.merge(row, Row(taskSubmitTime))
//                    row = Row.merge(row, Row(tenantId))
//                    row = Row.merge(row, Row(line.mkString("|")))
//                    row = Row.merge(row, Row(Array(s"$dbName|@|$tableName|@|$taskSubmitTime|@|${ef.field}|@|${line.getAs(fieldName)}|@|不符合转义字符检查规则")))
//                    rejectArray += row
//                  }
//                }
//              }
              case _ =>
            }
          }
        }
      })
      rejectArray
    })(encoder)
    rejectDF
  }

  def replaceWarnData(data: DataFrame, extractFields: List[TechCheckFieldInfo]): DataFrame = {

    val df: DataFrame = data

    val schemaFields = df.schema.fields
    val encoder = RowEncoder(df.schema)

    val replacedDF: DataFrame = df.map(line => {
      var row: Row = Row()

      extractFields.foreach(ef => {
        val fieldName = ef.field.toLowerCase
        val onOff = ef.option.tech_check.on_off
//        val ruleOptions = ef.option.tech_check.rule_options
        if (onOff && ef.data_type.equalsIgnoreCase("STRING")) {
          var defaultValue = line.getAs[String](fieldName)
          //替换警告数据中的乱码和转义字符
          for (rule_type <- CheckRuleConstants.ALL_RULE_TYPE) {
            rule_type match {
              case "MESSY_CODE" => {
                if (TechCheckRuleUtils.isMessyCode(line.getAs[String](fieldName))) {
                  defaultValue = TechCheckRuleUtils.replaceMessyCode(line.getAs[String](fieldName), CheckRuleConstants.MESSY_CODE_REPLACED_VALUE)
                }
              }
              case "ESCAPE_CHAR" => {
                if (TechCheckRuleUtils.isEscape(line.getAs[String](fieldName))) {
                  defaultValue = TechCheckRuleUtils.replaceEscape(line.getAs[String](fieldName), CheckRuleConstants.ESCAPE_CHAR_REPLACED_VALUE)
                }
              }
              case _ =>
            }
          }
          row = Row.merge(row, Row(defaultValue))
        } else {
          row = Row.merge(row, Row(line.getAs(fieldName)))
        }
      })
      row
    })(encoder)
    replacedDF
  }

  def checkSinkMQ(kc: KhaosContext, data: DataFrame): Unit = {

    //val nameServer = kc.conf.getString("khaos.quality.mq.nameServer")
    val nameServer = mqCheckInfo.getNamesrvAddr
    val topic = mqCheckInfo.getTopic
    val groupName = "diProducer"
    val tags = "dataIntegration"

    var df: DataFrame = data.repartition(1)
    df = df.select(concat_ws(",", df.columns.map(col_name => col(col_name)): _*) as "mq_sink")

//    log.info(s"CheckSinkMQDF schema is: ${df.printSchema}")
//    log.info(s"CheckSinkMQDF is: ${df.show(false)}")

    df.foreachPartition(partitionOfRecords => {
      val sinkList = new util.ArrayList[String]
      partitionOfRecords.foreach(info => {
        val mqSink = info.getAs[String]("mq_sink")
        sinkList.add(mqSink)
      })
      //log.info("===========khaos.quality.mq.nameServer========is:"+nameServer)
      log.info("开始上传检核结果到MQ")
      RocketMQProducer.syncSend(nameServer, groupName, topic, tags, sinkList)
    })
  }

  def emptyESDf(kc: KhaosContext): DataFrame = {
    val schema = StructType(Seq(StructField("taskFlowNum", StringType, true), StructField("ruleId", StringType, true),
      StructField("taskSubmitTime", StringType, true), StructField("tenantId", StringType, true),
      StructField("dataDetaile", StringType, true), StructField("errorDescribe", ArrayType(StringType), true)))

    val emptyRDD = kc.sparkSession.sparkContext.emptyRDD[Row]
    val emptyESDf = kc.sparkSession.createDataFrame(emptyRDD, schema)
    emptyESDf
  }

  def renameDF(data: DataFrame): DataFrame = {
    val colArr = new ArrayBuffer[Column]()
    data.schema.fields.foreach(field => {
      colArr += data.col(field.name) as (field.name.toLowerCase)
    })
    data.select(colArr: _*)
  }

  def timestamp2String(data: DataFrame): DataFrame = {
    val colArr = new ArrayBuffer[Column]()
    data.schema.fields.foreach(field => {
      if (field.dataType.equals(TimestampType)) {
        colArr += data.col(field.name).cast(StringType)
      } else {
        colArr += data.col(field.name)
      }
    })
    data.select(colArr: _*)
  }

  def transformDfSchema(targetDf: DataFrame, sourceDf: DataFrame): DataFrame = {
    var df: DataFrame = sourceDf
    targetDf.schema.fields.foreach(targetField => {
      sourceDf.schema.fields.foreach(sourceField => {
        if (sourceField.name.equals(targetField.name) && targetField.dataType.equals(TimestampType)) {
          df = df.withColumn(sourceField.name, df.col(sourceField.name).cast(TimestampType))
        }
      })
    })
    df
  }

  /** 将DF字段类型由String类型转为预期类型 */
  def convertDFType(kc: KhaosContext, data: DataFrame, schemaList: List[TechCheckFieldInfo]): DataFrame = {
    val columnArr = new ArrayBuffer[(String, String)]()
    for (i <- 0 until schemaList.size) {
      columnArr += (schemaList(i).field.toLowerCase -> schemaList(i).data_type)
    }

    val schema = SparkJobHelper.dynamicBuildDFSchema(columnArr)
    val rddArray: RDD[Array[String]] = data.rdd.map(row => row.toSeq.asInstanceOf[Seq[String]].toArray)

    val targetRDD = DataTypeConvertUtils.rdd2Rows(schema, rddArray)
    val targetDf = kc.sparkSession.createDataFrame(targetRDD, schema)
    targetDf
  }

  def timeStamp2Date(timestamp: String): String = {
    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    //要转换的时间格式
    val date = sdf.format(timestamp.toLong)
    date
  }

  def string2Date(date: String): String = {
    var formatter = new SimpleDateFormat("yyyyMMdd")
    formatter.setLenient(false)
    var newDate: Date = null
    newDate = formatter.parse(date)
    formatter = new SimpleDateFormat("yyyy-MM-dd")
    formatter.format(newDate)
  }
}